package com.atguigu.flink.charkoer13;

import com.atguigu.flink.been.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class FlinkTimeProcceing {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        DataStreamSource<WaterSensor> waterSensorStream =
                env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                        new WaterSensor("sensor_1", 2000L, 20),
                        new WaterSensor("sensor_2", 3000L, 30),
                        new WaterSensor("sensor_1", 4000L, 40),
                        new WaterSensor("sensor_1", 5000L, 50),
                        new WaterSensor("sensor_2", 6000L, 60)
                );


        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        tEnv.fromDataStream(waterSensorStream, $("ts"), $("id"), $("vc"), $("mafeifei").proctime()).execute().print();
        TableResult tableResult = tEnv.executeSql("create table abb(" +
                " id string," +
                " ts bigint," +
                " vc int," +
                " pt_time as PROCTIME()" +
                ")with(" +
                " 'connector' = 'filesystem' ," +
                " 'path' = 'input/sensor.txt' ," +
                " 'format' = 'csv' " +
                ")");

        tEnv.executeSql("select * from abb").print();
    }
}
